package cn.jbit.reactor;

import cn.jbit.reactor.worker.server.AsyncHandler;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

/**
 * 源码学院只为培养优秀的bat程序员而生
 *
 * @author 学员张在成
 * @data 2020/8/8
 */
public class Acceptor implements Runnable{

    //创建服务端通道对象
    private final ServerSocketChannel serverSocketChannel;
    //获取CPU核心数
    private  final  int coreNum=Runtime.getRuntime().availableProcessors();
    //轮询使用subReactor的下标索引
    private  int next=0;
    //创建Selector给subSelect使用,个数为CPU核心数
    private  final Selector[] selectors=new Selector[coreNum];

    private  SubReactor[] subReactors=new SubReactor[coreNum];

    private  Thread[] threads=new Thread[coreNum];

    public Acceptor(ServerSocketChannel serverSocketChannel) throws IOException {
        this.serverSocketChannel = serverSocketChannel;
        //初始化
        for (int i=0;i<coreNum;i++){
            selectors[i]=Selector.open();
            subReactors[i]=new SubReactor(selectors[i],i);
            threads[i]=new Thread(subReactors[i]);
            threads[i].start();
        }
    }

    @Override
    public void run() {
            SocketChannel socketChannel;
            try {
                while (true) {
                    socketChannel = serverSocketChannel.accept(); //阻塞获取连接
                    if (socketChannel!=null){
                        //轮询reactors处理接收到的请求
                        System.out.println(String.format("收到来自%s的连接",socketChannel.getRemoteAddress()));
                        socketChannel.configureBlocking(false);
                        //将当前的从反应堆设置为true,表示注册
                        subReactors[next].registering(true);
                        //多路复用器：将当前的多路复用所在的线程进行唤醒
                        selectors[next].wakeup();
                        SelectionKey selectionKey=socketChannel.register(selectors[next], SelectionKey.OP_READ);
                        selectors[next].wakeup();
                        subReactors[next].registering(false);
                        selectionKey.attach(new AsyncHandler(socketChannel, selectors[next]));

                        if (++next==selectors.length){
                            next=0;  //越界后重新分配
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
    }
}
